package org.example.snow.demo1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author snowsong
 */
public class StartRunApp {
    public static void main(String[] args) throws Exception {
        // 数据来源
        String inPath = "dataMsg/dataSourceFile.txt";
        // 数据输出
        String outputPath = "dataMsg/result.csv";
        // 初始化
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        // 读取文件信息
        DataStreamSource<String> textFile = executionEnvironment.readTextFile(inPath);
        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMap = textFile
                .flatMap(new Splitter())
                .keyBy(value -> value.f0)
                .sum(1);
        textFile.print();
        // 写入数据到 CSV 文件，使用 writeAsText 或 writeAsCsv
        flatMap.writeAsCsv(outputPath,FileSystem.WriteMode.OVERWRITE," ","\n")
                // 这是设置并行度的参数，表示执行该操作的任务的数量。在这里，setParallelism(1) 表示该操作会在 单个任务 中执行，而不是并行执行多个任务。通常情况下，Flink 在处理流或批数据时可以通过并行执行来加速处理，而设置并行度为 1 可以强制数据写入在一个线程中进行。
                .setParallelism(1);
        // 调用 execute() 后，Flink 会开始处理数据流中的每个操作，并根据定义的逻辑执行数据转换、聚合等操作。
        // 作业名称："file.txt -> result.csv" 是为了描述输入文件和输出文件的关系，可以帮助你理解这个作业的目的。这个名称在 Flink 的执行日志和监控界面中会显示
        executionEnvironment.execute("file.txt -> result.csv");

    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}
